<?php
namespace App\Services\Rabbitmq;

use Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMqService
{
    protected AMQPStreamConnection $connection;
    const QUEUE="MUNBERRMS";


    /**
     * @throws Exception
     */
    public function __construct()
    {
        $config = [
            'host' => env("RABBITMQ_HOME"),
            'port' => env("RABBITMQ_PORT"),
            'user' => env("RABBITMQ_USER"),
            'password' => env("RABBITMQ_PASS"),
            'vhost' => env("RABBITMQ_VHOST"),//MunberRms虚拟主机
        ];
        return $this->connection = new AMQPStreamConnection($config['host'], $config['port'], $config['user'], $config['password'], $config['vhost']);
    }


    /*
     * 数据插入到mq队列中（生产者）
     * @param $queue .队列名称
     * @param $messageBody .消息体
     * @param string $exchange .交换机名称
     * @param string $routing_key .设置路由
     * @param string $class .消息接受对象类型
     * @throws \Exception
     */
    public function push($routing_key, $messageBody,$class = 1):string
    {
        //$class = 1;
        //class 接受对象:1自己，2司机，3货主，4所有人

        $queue=self::QUEUE;
        $Noti = (array)json_decode($messageBody);
        $Message_types = [
            'app_auth'=>[
                "name_ug"=>"سىېستىما ئۇچۇرى",
                "name_zh"=>"系统通知"
            ],
            'company_auth'=>[
                "name_ug"=>"كارخانا ئۇچۇرى",
                "name_zh"=>"集团-公司通知"
            ],
            'vip_auth'=>[
                "name_ug"=>"ئەزالىق ئۇچۇرى",
                "name_zh"=>"会员"
            ],
            'order_create'=>[
                "name_ug"=>"تالۇن ئۇچۇرى",
                "name_zh"=>"订单消息"
            ],
            'order_seize'=>[
                "name_ug"=>"تالۇن قوبۇللاش ئۇچۇرى",
                "name_zh"=>"接单通知"
            ],
            'person_auth'=>[
                "name_ug"=>"شەخىسىي ئۇچۇر",
                "name_zh"=>"个人消息"
            ]
        ];

        $this->Socket($queue,$routing_key,$messageBody,$Noti,$Message_types,$class);
        return '生产者已操作';

    }
    private function Socket($queue,$routing_key,$messageBody,$Noti,$Message_types,$class): void
    {
        #原理:
        /*
        RabbitMQ 里面的消息是从
        1.生产者(Producer) 发布消息
        2.交换机
        3.队列(消息的存储器)
        4.队列和交换机绑定
        5.消费者接受
        */
        //构建通道（mq的数据存储与获取是通过通道进行数据传输的）


        //适合多个队列和单个交换机
        $channel = $this->connection->channel();

        $exchange = "MUNBERRMS";

        $queue .= "_".$routing_key;

        //指定交换机，若是路由的名称不匹配不会把数据放入队列中
        $channel->exchange_declare($exchange, 'direct', false, true, false);//direct ->第二个参数；

        //声明一个队列
        $channel->queue_declare($queue,false, true, false, true);//

        //队列和交换器绑定/绑定队列和类型
        $channel->queue_bind($queue, $exchange,(int)$routing_key);




        $config = [
            'content_type' => 'text/plain',
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
        ];

        //实例化消息推送类
        $message = new AMQPMessage($messageBody, $config);

        //消息推送到路由名称为$exchange的队列当中

        $channel->basic_publish($message, $exchange,(int)$routing_key); //



        //关闭消息推送资源
        $channel->close();
        //关闭mq资源
        $this->connection->close();
    }

    /**
     * 消费者：取出消息进行消费，并返回
     * @param $queue //队列名称
     * @param $callback //回调函数,处理队列里的消息
     * @return bool
     * @throws Exception
     */
    public function pop($queue, $callback)
    {
        echo '消费者中心' . PHP_EOL;
        //连接到 RabbitMQ 服务器并打开通道
        $channel = $this->connection->channel();
        //声明要获取内容的队列
        $channel->queue_declare($queue, false, true, false, false);
        //获取队列信息(消息总条数)
        $countQueue = $channel->queue_declare($queue, true)[1];
        //获取队列中的下一条消息
        $msg = $channel->basic_get($queue);
        //确认消费消息
        $channel->basic_ack($msg->delivery_info['delivery_tag']);//消息主题返回给回调函数

        //消息主题返回给回调函数
        $res = $callback($msg->body);
        if ($res) {
            echo 'ack验证' . PHP_EOL;
            //ack验证，如果消费失败了，从新获取一次数据再次消费
            $channel->basic_ack($msg->getDeliveryTag());
        }
        echo 'ack消费完成' . PHP_EOL;
        $channel->close();
        $this->connection->close();
        return true;
    }

}
